package com.jarveis.frame.redis.mq;

import java.util.List;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

/**
 * @desc 消息队列
 * @author liuguojun
 * @date 2016-01-14
 */
public class RMQQueue {

	private static final Logger logger = Logger.getLogger(RMQQueue.class);

	public static final String QUEUE = "queue:";
	private String key;
	private Jedis jedis;

	public RMQQueue(String key, Jedis jedis) {
		this.key = QUEUE + key;
		this.jedis = jedis;
	}

	/**
	 * 队列标识
	 * 
	 * @return
	 */
	public String key() {
		return this.key;
	}

	/**
	 * 开启事务
	 * 
	 * @return
	 */
	public Transaction transaction() {
		jedis.watch(key());
		return jedis.multi();
	}

	/**
	 * 添加消息
	 * 
	 * @param message
	 */
	public void publish(final String message) {
		long num = 0;
		do {
			num = jedis.rpush(key(), message);
			logger.info("jedis.rpush(" + key() + ", " + message + ")=" + num);
		} while (num < 1);
	}

	/**
	 * 添加消息
	 * 
	 * @param message
	 * @param istrans
	 * @deprecated
	 */
	public void publish(final String message, boolean istrans) {
		if (!istrans) {
			publish(message);
		} else {
			List<Object> exec = null;
			do {
				Transaction trans = transaction();
				trans.rpush(key(), message);
				exec = trans.exec();
				logger.info("trans.rpush(" + key() + ", " + message + ")="
						+ exec);
			} while (exec == null);
		}
	}

	/**
	 * 消费消息
	 * 
	 * @param callback
	 */
	public void consume(RMQCallback callback) {
		while (true) {
			String message = read();
			if (message != null) {
				try {
					callback.onMessage(message);
				} catch (Exception e) {
					// 如果数据处理异常，再重新加入到队列中
					publish(message);
				}
			} else {
				waitting();
			}
		}
	}

	/**
	 * 消费消息
	 * 
	 * @return
	 */
	public String consume() {
		return read();
	}

	/**
	 * 获取未读的消息数
	 * 
	 * @return
	 */
	public long count() {
		return jedis.llen(key());
	}

	/**
	 * 读取消息
	 * 
	 * @return
	 */
	private String read() {
		while (count() > 0) {
			String message = jedis.lpop(key());
			if (message != null) {
				return message;
			}
		}

		return null;
	}

	/**
	 * 等待消息
	 */
	private void waitting() {
		try {
			Thread.sleep(500);
		} catch (InterruptedException ex) {
			logger.error(ex.getMessage(), ex);
		}
	}

}